package com.asura.ops.sync.sdk;

import cn.hutool.json.JSONUtil;
import com.asura.ops.sync.api.SyncApi;
import com.asura.ops.sync.api.model.ErrorReportDto;
import com.asura.ops.sync.sdk.consumer.ChangeConsumer;
import com.asura.ops.sync.sdk.consumer.Result;
import com.asura.ops.sync.sdk.model.MqPayload;
import com.asura.ops.sync.sdk.properties.CfgClientProperties;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.support.TransactionTemplate;

import java.nio.charset.StandardCharsets;

/**
 * @author: zouyang
 * @date: 2022/8/4
 * @description: 同步消息监听
 */
@Slf4j
public class SyncMessageListener extends MessageListenerAdapter {
    private static String TABLE = "$table$";
    private static String UPDATE_REPLACE_DATA = "$replaceData$";
    private static String UPDATE = "UPDATE " + TABLE + " SET " + UPDATE_REPLACE_DATA + " WHERE id=?";
    private static String INSERT_COLUMN = "$COLUMNS$";
    private static String INSERT_VAL = "$INSERT_VAL$";
    private static String DELETE = "DELETE FROM " + TABLE + " WHERE id=?";
    private static String INSERT = "INSERT INTO " + TABLE + "(" + INSERT_COLUMN + ") VALUES(" + INSERT_VAL + ")";
    private JdbcTemplate jdbcTemplate;
    private TransactionTemplate transactionTemplate;

    public SyncMessageListener() {
    }

    public SyncMessageListener(JdbcTemplate jdbcTemplate, TransactionTemplate transactionTemplate) {
        this.jdbcTemplate = jdbcTemplate;
        this.transactionTemplate = transactionTemplate;
    }

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        String msg = new String(message.getBody(), StandardCharsets.UTF_8);

        MqPayload payload = JSONUtil.toBean(msg, MqPayload.class);

        if (payload == null) {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }

        log.info("接收到消息:{}", payload.getMsgId());

        ChangeConsumer changeConsumer = ApplicationContextUtil.getObject(ChangeConsumer.class);
        Result result = changeConsumer.handle(payload);
        if (result.isSuccess()) {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } else {
            log.error("消息:{} 处理失败", payload.getMsgId());
            //上报处理异常消息
            ErrorReportDto errorReport = new ErrorReportDto();
            errorReport.setErrorMsg(result.getErrorMsg());
            errorReport.setClientCode(ApplicationContextUtil.getObject(CfgClientProperties.class).getClientCode());
            errorReport.setDbName(payload.getSource().getDb());
            errorReport.setTableName(payload.getSource().getTable());
            errorReport.setMsgId(payload.getMsgId());

            ApplicationContextUtil.getObject(SyncApi.class).errorReport(errorReport);

            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }

//        transactionTemplate.execute(status -> {
//            try {
//                String table = payload.getSource().getTable();
//                Integer type = payload.getChangeType();
//                String sql = "";
//                String id = "id";
//                switch (type) {
//                    case 1:
//                        Set<String> keySet = payload.getAfter().keySet();
//                        List<String> fields = Lists.newArrayList();
//                        List<Object> values = Lists.newArrayList();
//                        for (String field : keySet) {
//                            fields.add(field);
//                            if (field.equals("created_at") || field.equals("updated_at")) {
//                            } else {
//                                values.add(payload.getAfter().get(field));
//                            }
//                        }
//                        String insertColumn = StrUtil.join(",", fields);
//                        String insertVals = StrUtil.repeatAndJoin("?", fields.size(), ",");
//                        sql = INSERT.replace(TABLE, table).replace(INSERT_COLUMN, insertColumn).replace(INSERT_VAL, insertVals);
//                        jdbcTemplate.update(sql, values.toArray());
//                        break;
//                    case 2:
//                        Set<String> keySet2 = payload.getAfter().keySet();
//                        String join = StrUtil.join("=?,", keySet2);
//                        String replaceData = join + "=?";
//                        sql = UPDATE.replace(TABLE, table).replace(UPDATE_REPLACE_DATA, replaceData);
//                        jdbcTemplate.update(sql, payload.getAfter().values(), payload.getAfter().get(id));
//                        break;
//                    case 3:
//                        sql = DELETE.replace(TABLE, table);
//                        jdbcTemplate.update(sql, payload.getBefore().get(id));
//                        break;
//                }
//                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//                return true;
//            } catch (Exception exception) {
//                exception.printStackTrace();
//                try {
//                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//                //TODO 处理消费失败的消息
//                return false;
//            }
//        });

    }
}
